package com.hl233.framework.utils;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.hl233.framework.entity.KafkaDataEntity;
import com.hl233.framework.mapper.KafkaData1Mapper;
import org.springframework.beans.factory.annotation.Autowired;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author 张振强15065206763
 * @create 2024-09-06
 */
public class KafkaDataConsumer1 implements Runnable {

    @Autowired
    private KafkaData1Mapper kafkaData1Mapper;

    @Autowired
    private DataSource dataSource;
    @Autowired
    private DruidDataSource druidDataSource;

    @Autowired
//    DruidPooledConnection druidPooledConnection = new DruidPooledConnection();

    private final String name;

    public KafkaDataConsumer1(String name) {
        this.name = name;
    }


    @Override
    public void run() {
        while (true) {
            KafkaDataEntity bread = KafkaDataQueue.take();
            System.out.println("拿到这个东西"+bread);
            Connection conn = null;
            PreparedStatement pstmt = null;
            try {
                conn = druidDataSource.getConnection();
                String sql = "INSERT INTO kafka_data_1 (context, createtime, val) VALUES (?, ?, ?)";
                pstmt = conn.prepareStatement(sql);
                pstmt.setString(1, bread.getContext());
                pstmt.setString(2, bread.getCreatetime());
                pstmt.setInt(3, bread.getVal());
                pstmt.execute();
                System.out.println(name + "购买了" + bread + ":" + "面包");
            } catch (SQLException e) {
                try {
                    if (conn != null) {
                        conn.rollback(); // 在发生异常时回滚事务
                    }
                } catch (SQLException ex) {
                    ex.printStackTrace();
                }
                e.printStackTrace();
            } finally {
                // 关闭资源
                try {
                    if (pstmt != null) pstmt.close();
                    if (conn != null) conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
